use crossbeam::channel::{self, Sender};
use log::{debug, info};
use std::io::{BufWriter, Write};
use std::error::Error;
use std::result::Result;
use std::{
    io::BufReader,
    net::{SocketAddr, TcpListener, TcpStream},
    str::FromStr,
};

use crate::error::CustomError;
use crate::execute::execute::execute;
use crate::sql::parser::Lexer;
use crate::storage::bitcash::BitCash;
use crate::storage::{BitCashEngine, Engine, SendMessage};

use super::{Request, Response};

pub struct Serve {
    pub socket_addr: SocketAddr,
    pub port: u16,
}

impl Serve {
    pub fn new(ip: &str, port: u16) -> Self {
        let socket_addr = match SocketAddr::from_str(&format!("{}:{}", ip, port)) {
            Ok(ipv4_addr) => ipv4_addr,
            Err(e) => {
                panic!("IPv4 地址解析失败: {}", e);
            }
        };
        Serve { socket_addr, port }
    }

    pub async fn start(&self) -> Result<(), Box<dyn Error>> {
        let listener = TcpListener::bind(self.socket_addr)?;
        let (sender, receiver) = channel::unbounded::<SendMessage>();

        tokio::spawn(async move {
            let bc = BitCash::new("/tmp/rust-db/", "db.log");
            let mut engine = Box::new(BitCashEngine::new(bc, receiver));
            let _ = engine.execute();
        });

        while let Ok(stream_result) = listener.accept() {
            let stream = stream_result.0;
            let sender_clone = sender.clone();
            tokio::spawn(async move {
                let _ = handle_client(stream, sender_clone).await;
            });
        }
        Ok(())
    }
}

pub async fn handle_client(tcp_stream: TcpStream, sender: Sender<SendMessage>) -> Result<(), Box<dyn Error>> {
    let mut reader = BufReader::new(tcp_stream.try_clone()?);
    let mut writer = BufWriter::new(tcp_stream);

    while let Ok(Some(request)) = reader_from_socket(&mut reader) {
        debug!("message: {:#?}", request);
        match request {
            Request::Execute(sql) => {
                let mut lexer = Lexer::new(&sql);
                let token = lexer.parse()?;
                let response = execute(token, sender.clone())?;
                debug!("response: {:#?}", response);
                write_to_socket(&mut writer, response)?;
            },
            Request::GetTable(sql) => {},
            Request::ListTables => {},
            Request::Status => {}
        }
        writer.flush()?;
    }

    Ok(())
}

fn write_to_socket<W>(writer: W, value: Response) -> Result<(), Box<dyn Error>> where W: std::io::Write {
    bincode::serialize_into(writer, &value)?;
    Ok(())
}

fn reader_from_socket(
    reader: &mut BufReader<TcpStream>,
) -> Result<Option<Request>, Box<dyn Error>> {
    let request: Option<Request> = match bincode::deserialize_from(reader) {
        Ok(v) => Some(v),
        Err(e) => match *e {
            bincode::ErrorKind::Io(e) => match e.kind() {
                std::io::ErrorKind::UnexpectedEof => None,
                std::io::ErrorKind::ConnectionReset => None,
                _ => {
                    return Err(
                        CustomError::SocketReadError("IO read socket error".to_string()).into(),
                    )
                }
            },
            _ => {
                return Err(
                    CustomError::SocketReadError("NetWork read socket error".to_string()).into(),
                )
            }
        },
    };
    debug!("{:?}", request);
    Ok(request)
}
